<?php
// https://github.com/videlalvaro/php-amqplib
require_once __DIR__.'/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$connect = new AMQPStreamConnection('192.168.205.128', 5672, 'admin', '123456');
$channel = $connect->channel();
$channel->queue_declare('task_queue', false, true, false, false);

echo ' [*] Waiting for messages. To exit press CTRL+C'.PHP_EOL;

$callbackFunction = function ($msg) {
    $deliveryTag = $msg->delivery_info['delivery_tag'];
    echo "[x] Received Tag: $deliveryTag Body：". $msg->body.PHP_EOL;
    sleep( substr_count($msg->body, '.') );
    echo "[x] Finish ".date('Y-m-d H:i:s').PHP_EOL.PHP_EOL;
    $msg->delivery_info['channel']->basic_ack($deliveryTag);    // 发送确认消息
};

// 公平调度，在确认前一个消息之前，不要向消费者发送新的消息(一次给消费者发送一个消息)
$channel->basic_qos(null, 1, null);
// 开启消息确认机制
$channel->basic_consume('task_queue', '', false, false, false, false, $callbackFunction);

while (count($channel->callbacks)) {
    $channel->wait();
}

$channel->close();
$connect->close();